package org.example
import org.apache.spark.sql.SparkSession
object spark_data2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .getOrCreate()
    val sc = spark.sparkContext
    val filepath = "src/main/java/com/movie/data/"
    val usersRDD = sc.textFile(filepath + "users.dat")
    val occupationsRDD = sc.textFile(filepath + "occupations.dat")
    val ratingsRDD = sc.textFile(filepath + "ratings.dat")
    val moviesRDD = sc.textFile(filepath + "movies.dat")
    occupationsRDD.cache()
    usersRDD.cache()
    ratingsRDD.cache()
    moviesRDD.cache()
    println("职业数:" + occupationsRDD.count())
    println("电影数:" + moviesRDD.count())
    println("用户数:" + usersRDD.count())
    println("评分条数:" + ratingsRDD.count)


    val userId = "20"
    //"ratings.dat"：UserID::MovieID::Rating::Timestamp //  方式一
    //   val userWatchedMovie = usersRDD.map(x => x.split("::")).
    //   filter( //item => item(0).equals(userId) //   )
    //  println(userId + "观看过的电影数: " + userWatchedMovie.count())
    // 方式二
    //   val userWatchMovie = ratingsRDD.map(x => x.split("::")) //
    //   .map(user => (user(1), user(0))) //  .filter(_._2.equals(userId))
    // 157
    val userWatchMovie = ratingsRDD.map(x => x.split("::"))
      //(MovieID,UserId)
      .map(user => (user(1), user(0)))
      .filter(_._2.equals(userId))
    // userid count // 1  53 // 20   24 // 18   305
    // val data = userWatchMovie.lookup(userId)  println(userId + " \n 观看过的电影数: \n" + userWatchMovie.count())
    println(userId + "  观看过的电影数: " + userWatchMovie.count())
    println("这些电影的详情: \n")
    // MovieID::Title::Genres
    val movieInfoRDD = moviesRDD.map(_.split("::"))
      //  [MovieID,Title,Genres]
      .map(movie => {
        (movie(0), (movie(1), movie(2)))
      })
    //  (  MovieId, (Title,Genres) )
    val result = userWatchMovie.join(movieInfoRDD)
      // (MovieId,(UserId, (Title,Genres) ))
      .map(item => {
        (item._1, item._2._1, item._2._2._1, item._2._2._2)
      })
    result.take(10).foreach(println)

    //需求2
    // "ratings.dat"：UserID::MovieID::Rating::Timestamp
    // [UserID,MovieID,Rating,Timestamp]
    val rating = ratingsRDD.map(x => x.split("::")).map {
      x => {
        (x(0), x(1), x(2)) // (UserID,MovieID,Rating)
      }
    }.cache()
    println("平均得分最高的前 10 名的电影名称简单版")
    //  ( MovieId,( Rating, 1) )
    //  (1200,(4.0,1))
    rating.map(x => (x._2, (x._3.toDouble, 1)))
      //  ( MovieId,( 总分, 总次数) )
      .reduceByKey((x, y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      // (  平均分, MovieId )
      .map(x => (x._2._1 / x._2._2, x._1))
      .sortByKey(false)
      .take(5)
      .foreach(println)
    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次数))")
    // MovieID::Title::Genres
    val moviesInfo = moviesRDD.map(x => x.split("::"))
      .map(x => {
        (x(0), (x(1), x(2)))
      })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
      //  ( MovieId,( Rating, 1) )
      //  (1252,(4.0,1))
      .reduceByKey((x, y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      // ( MovieId,( 总分, 总次数) )
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
    //( MovieId, (平均分, 总分,总次数) )
    moviesInfo.join(ratingsInfo)
      .map(info => {
        (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
        // (平均分,(movieId,Title,Genres,总分,总次数))  }).sortByKey(false)
        // .take(5)
        // .foreach(println)
      }).sortByKey(false)
      .take(5)
      .foreach(println)
    println("观影人数最多的前10部电影")
    val watchViewsInfo = rating.map(x => {
        (x._2, 1)
      }).reduceByKey((x, y) => x + y)
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(5)
    watchViewsInfo.foreach(println(_))
    println("=====================")
    rating.map(x => (x._2, 1))
      .reduceByKey((x, y) => {
        (x + y)
      })
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
      .foreach(println)
    println("详情的输出(  观影人数，电影编号)")
    moviesInfo.join(ratingsInfo).map(x => {
        (x._2._2._3, (x._2._1._1, x._2._1._2, x._2._2._1, x._2._2._1))
      }).sortByKey(false)
      .take(10)
      .foreach(println)
    sc.stop()
  }
}
